package com.peng.sparktest.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object SparkApiTest_TOPN2 {


  implicit val myOrder = new Ordering[(Int, Int)] {
    override def compare(x: (Int, Int), y: (Int, Int)) = y._2.compareTo(x._2)
  }


  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setAppName("topN").setMaster("local")
    val context = new SparkContext(conf)
    context.setLogLevel("ERROR")
    val file = context.textFile("test_file/topnfile")
    //    样例数据：1970-8-8	32
    //    val mapped = file.map(line => {
    //      val strings = line.split("\t")
    //      val ymdStrs = strings(0).split("-")
    //      (ymdStrs(0).toInt, ymdStrs(1).toInt, ymdStrs(2).toInt, strings(1).toInt)
    //    })


    // 3、 用了groupByKey  容易OOM  取巧：用了spark 的RDD 的reduceByKey 去重，用了sortByKey 排序
    /**
     * 多级shuffle  后续shuffle参与分区的key一定得是前置分区 key的子集  ，才可以保证顺序性
     */
    // 有问题的版本:
    //    val sorted: RDD[(Int, Int, Int, Int)] = mapped.sortBy(t4 => (t4._1, t4._2, t4._4), false)
    //    val reduced: RDD[((Int, Int, Int), Int)] = sorted.map(t4 => ((t4._1, t4._2, t4._3), t4._4)).reduceByKey((x: Int, y: Int) => if (y > x) y else x)
    //    val maped: RDD[((Int, Int), (Int, Int))] = reduced.map(t2 => ((t2._1._1, t2._1._2), (t2._1._3, t2._2)))
    //    val grouped: RDD[((Int, Int), Iterable[(Int, Int)])] = maped.groupByKey()
    //    grouped.foreach(println)

    // 正确案例：
    //    val sorted: RDD[(Int, Int, Int, Int)] = mapped.sortBy(t4 => (t4._1, t4._2, t4._4), false)
    //    //    val reduced: RDD[((Int, Int, Int), Int)] = sorted.map(t4 => ((t4._1, t4._2, t4._3), t4._4)).reduceByKey((x: Int, y: Int) => if (y > x) y else x)
    //    val maped = sorted.map(t4 => ((t4._1, t4._2), (t4._3, t4._4)))
    //    val grouped = maped.groupByKey()
    //
    //    grouped.mapValues(values => {
    //      val map = new mutable.HashMap[Int, Int]()
    //      values.foreach(item => if (map.getOrElse(item._1, 0) < item._2) map.put(item._1, item._2))
    //      map.toList
    //    }).foreach(println)

    /**
     * 分布式计算的核心思想：调优天下无敌：自定义的combineByKey
     * 此方案完美解决OOM
     */
    val mapped: RDD[((Int, Int), (Int, Int))] = file.map(line => {
      val strings = line.split("\t")
      val ymdStrs = strings(0).split("-")
      ((ymdStrs(0).toInt, ymdStrs(1).toInt), (ymdStrs(2).toInt, strings(1).toInt))
    })

    val resultRes = mapped.combineByKey( //同年同月

      (item: (Int, Int)) => mutable.Map(item._1 -> item._2),
      (oldV: mutable.Map[Int, Int], newV: (Int, Int)) => {
        var flag = true
        for ((k, v) <- oldV if flag) {
          if (v < newV._2 || oldV.size < 2) {
            {
              if (oldV.getOrElse(newV._1, -1) < newV._2) {
                oldV.+=(newV)
                if (oldV.size > 2) {
                  oldV.remove(oldV.min._1)
                }
                flag = false

              }

            }

          }
        }
        oldV
      },
      (oldV: mutable.Map[Int, Int], newV: mutable.Map[Int, Int]) => {
        oldV.++(newV)
      }

      // combine合并的时候也需要对同一天去重，例如一边是1号11度，1边是1号22度，应该是只留下1号22度，留下同一天中最高的温度
      //      (oldV: Array[(Int, Int)], newV: Array[(Int, Int)]) => {
      //        var addNums = Seq.newBuilder[(Int, Int)]
      //        for (newI <- newV.indices) {
      //          var flag = 0
      //          for (oldI <- oldV.indices) {
      //            if (oldV(oldI)._1 == newV(newI)._1) {
      //              if (oldV(oldI)._2 < newV(newI)._2) {
      //                oldV(oldI) = newV(newI)
      //              }
      //              flag = 1
      //            }
      //            if (flag == 0) {
      //              addNums.+=(newV(newI))
      //            }
      //          }
      //        }
      //        val result: Array[(Int, Int)] = oldV.union(addNums.result())
      //        util.Sorting.quickSort(result)
      //        result
      //
      //      }
    )
    resultRes.mapValues(item => item.toList.sorted).foreach(println)

    //    resultRes
    //      .mapValues(item => item.filter(one => one != (0, 0)).take(2))
    //      .map((item: ((Int, Int), Array[(Int, Int)])) => (item._1, item._2.toList)).foreach(println)


    //    ((2018,3),List((11,18), (0,0), (0,0)))
    //    ((2019,5),List((21,33), (0,0), (0,0)))
    //    ((2018,4),List((23,22), (0,0), (0,0)))
    //    ((1970,8),List((0,0), (23,23), (8,32)))
    //    ((2019,6),List((0,0), (0,0), (3,40)))


    //    ((2018,3),List((11,18), (0,0), (0,0)))
    //    ((2019,5),List((21,33), (0,0), (0,0)))
    //    ((2018,4),List((23,22), (0,0), (0,0)))
    //    ((1970,8),List((0,0), (23,23), (8,32)))
    //    ((2019,6),List((0,0), (0,0), (3,40)))


    //    ((2018,3),List((11,18)))
    //    ((2019,5),List((21,33)))
    //    ((2018,4),List((23,22)))
    //    ((1970,8),List((8,32), (23,23)))
    //    ((2019,6),List((3,40), (1,39), (2,31)))

    //    ((2018,3),List((11,18), (0,0), (0,0)))
    //    ((2019,5),List((21,33), (0,0), (0,0)))
    //    ((2018,4),List((23,22), (0,0), (0,0)))
    //    ((1970,8),List((0,0), (23,23), (8,32)))
    //    ((2019,6),List((0,0), (0,0), (3,40)))

    while (true) {

    }


  }

}
